package com.lagou.hwhbase.processor;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellScanner;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver;
import org.apache.hadoop.hbase.coprocessor.ObserverContext;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.List;

public class HwProcessor extends BaseRegionObserver {
    @Override
    public void postDelete(ObserverContext<RegionCoprocessorEnvironment> e, Delete delete, WALEdit edit, Durability durability) throws IOException {
        System.out.println("****************开始执行协处理器代码****************");
        try {
            RegionCoprocessorEnvironment env = e.getEnvironment();
            List<Cell> cells = delete.getFamilyCellMap().firstEntry().getValue();
            Cell cell = cells.get(0);
            //从cell中获取rowKey 即对方的qualifier
            String rowkey = Bytes.toString(CellUtil.cloneRow(cell));
            //从cell中获取列名 即对方的rowkey
            String qualifier = Bytes.toString(CellUtil.cloneQualifier(cell));
            //获取表
            Table hwTable = env.getTable(TableName.valueOf("hw"));
            Get get = new Get(Bytes.toBytes(qualifier));
            get.addColumn(Bytes.toBytes("friends"), Bytes.toBytes(rowkey));
            get.setCheckExistenceOnly(true);
            Result result = hwTable.get(get);
            if (result.getExists()) {//存在则删除,避免死循环
                Delete delete1 = new Delete(Bytes.toBytes(qualifier));
                delete1.addColumn(Bytes.toBytes("friends"), Bytes.toBytes(rowkey));
                hwTable.delete(delete1);
            }
        } catch (Exception ex) {
            System.out.println(ex.getMessage());
        }
    }
}
